package com.lcf.user.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

@Component
public class MyConsumer {
    private static Logger logger = LoggerFactory.getLogger(MyProducer.class);

    /**
     * 指定topic
     *
     * @param record
     */
    @KafkaListener(topics = ("first-topic"))
    public void listen1(ConsumerRecord<?, ?> record) {
        Optional<?> value = Optional.ofNullable(record.value());
        if (value.isPresent()) {
            String o = (String) value.get();
            logger.info("获取到的内容：" + o);
        }
    }

    /**
     * 指定分组和topic
     *
     * @param record
     */
    @KafkaListener(groupId = "aa", topics = ("first-topic"))
    public void listen2(ConsumerRecord<?, ?> record) {
        String key = (String) record.key();
        Optional<?> value = Optional.ofNullable(record.value());
        int partition = record.partition();
        if (value.isPresent()) {
            String o = (String) value.get();
            logger.info("aa获取到的内容：" + o);
            logger.info("aa获取到的key内容：" + key);
            logger.info("aa获取到的partition内容：" + partition);
        }
    }

    /**
     * 指定接收的分区
     *
     * @param record
     */
    @KafkaListener(id = "id0", groupId = "bb", topicPartitions = {@TopicPartition(topic = "first-topic", partitions = {"0", "1"})})
    public void listen3(ConsumerRecord<?, ?> record) {
        String key = (String) record.key();
        int partition = record.partition();
        Optional<?> value = Optional.ofNullable(record.value());
        if (value.isPresent()) {
            String o = (String) value.get();
            logger.info("bb获取到的内容：" + o);
            logger.info("bb获取到的key内容：" + key);
            logger.info("bb获取到的partition内容：" + partition);
        }
    }

    /**
     * 监听topic
     *
     * @param messageList 消息
     * @param ack         手动提交对象
     */
    @KafkaListener(topics = "aa", containerFactory = "listenerAckFactory")
    public void listenOther(List<ConsumerRecord<?, ?>> messageList, Acknowledgment ack) {
        logger.info("本次监听到数据量：{}", messageList.size());
        try {

        } catch (Exception e) {

        } finally {
            ack.acknowledge();
        }
    }
}
